package cn.doitedu.rtdw.dataetl;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB04_视频播放分析OLAP支撑 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建kafka明细宽表的逻辑映射表

        // 建表，映射 kafka中 dwd层 行为日志明细宽表
        tenv.executeSql(
                " create table dwd_events_kafka (                       "+
                        "     release_channel string                             "+
                        "     ,device_type string                                 "+
                        "     ,session_id  string                                 "+
                        "     ,lat         double                                 "+
                        "     ,lng         double                                 "+
                        "     ,username    string                                 "+
                        "     ,event_id     string                                "+
                        "     ,action_time  bigint                                "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        " 	  ,member_level_id bigint                             "+
                        "     ,password string                                    "+
                        "     ,nickname string                                    "+
                        "     ,phone string                                       "+
                        "     ,status int                                         "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,icon string                                        "+
                        "     ,gender int                                         "+
                        "     ,birthday date                                      "+
                        "     ,register_city string                               "+
                        "     ,job string                                         "+
                        "     ,personalized_signature string                      "+
                        "     ,source_type int                                    "+
                        "     ,integration int                                    "+
                        "     ,growth int                                         "+
                        "     ,lucky_count int                                    "+
                        "     ,history_integration int                            "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,page_lanmu string                                  "+
                        "     ,page_pindao string                                 "+
                        "     ,pt AS proctime()                                   "+   // 处理时间语义
                        "     ,rt AS to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+   // 事件时间语义
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    "+
                        "     'properties.group.id' = 'goo2',                     "+
                        "     'scan.startup.mode' = 'latest-offset',              "+
                        "     'value.format'='json',                              "+
                        "     'value.json.fail-on-missing-field'='false',         "+
                        "     'value.fields-include' = 'EXCEPT_KEY'               "+
                        " )                                                       "
        );


        // sql统计
        // 滚动窗口内，group by聚合
        tenv.executeSql(
                "CREATE TEMPORARY VIEW agg_res AS  " +
                        "WITH tmp1 AS (                                                                                              "+
                        "     SELECT                                                                                                  "+
                        "         user_id,                                                                                            "+
                        "     	province,                                                                                             "+
                        "     	device_type,                                                                                          "+
                        "     	cast(properties['video_id'] as int) as video_id,                                                                   "+
                        "     	properties['play_id'] as play_id,                                                                     "+
                        "     	action_time,                                                                                          "+
                        " 		rt,                                                                                                   "+
                        "     	if(event_id='video_resume',1,0) as flag                                                               "+
                        "     FROM dwd_events_kafka                                                                                   "+
                        "     WHERE event_id in ('video_play','video_hb','video_pause','video_resume','video_stop')                   "+
                        " )                                                                                                           "+
                        " ,tmp2 AS (                                                                                                  "+
                        "     SELECT                                                                                                  "+
                        "         user_id,                                                                                            "+
                        "         province,                                                                                           "+
                        "         device_type,                                                                                        "+
                        "         video_id,                                                                                           "+
                        "         play_id,                                                                                            "+
                        "         action_time,                                                                                        "+
                        "         sum(flag) over(partition by play_id order by rt ) as flag,                                          "+
                        " 		rt                                                                                                    "+
                        "     FROM tmp1                                                                                               "+
                        " )                                                                                                           "+
                        "                                                                                                             "+
                        " ,tmp3 AS (                                                                                                  "+
                        "     SELECT                                                                                                  "+
                        "         user_id,                                                                                            "+
                        "         province,                                                                                           "+
                        "         device_type,                                                                                        "+
                        "         video_id,                                                                                           "+
                        "         play_id,                                                                                            "+
                        "         action_time,                                                                                        "+
                        "         flag,                                                                                               "+
                        " 		first_value(action_time) over(partition by play_id,flag order by rt) as start_time,                   "+
                        " 		rt                                                                                                    "+
                        "     FROM tmp2                                                                                               "+
                        " )                                                                                                           "+
                        "                                                                                                             "+
                        " SELECT                                                                                                      "+
                        "     user_id,                                                                                                "+
                        "     province,                                                                                               "+
                        "     device_type,                                                                                            "+
                        "     video_id,                                                                                               "+
                        "     play_id,                                                                                                "+
                        "     flag,                                                                                                   "+
                        " 	  start_time,                                                                                             "+
                        " 	  max(action_time) as end_time,                                                                           "+
                        " 	  proctime() as pt                                                                                        "+
                        " FROM TABLE(                                                                                                 "+
                        "     TUMBLE(TABLE tmp3,DESCRIPTOR(rt),INTERVAL '1' MINUTE)                                                   "+
                        " )                                                                                                           "+
                        " GROUP BY                                                                                                    "+
                        "     window_start,                                                                                           "+
                        " 	window_end,                                                                                               "+
                        "     user_id,                                                                                                "+
                        "     province,                                                                                               "+
                        "     device_type,                                                                                            "+
                        "     video_id,                                                                                               "+
                        "     play_id,                                                                                                "+
                        "     flag,                                                                                                   "+
                        " 	start_time                                                                                                "
        );


        // 建doris物理表的逻辑映射表
        tenv.executeSql(
                " CREATE TABLE video_play_agg_doris(         "+
                        "   user_id bigint,                              "+
                        " 	province varchar(20),                        "+
                        " 	device_type varchar(20),                     "+
                        " 	video_id int,                                "+
                        " 	video_name varchar(40),                      "+
                        " 	video_type varchar(20),                      "+
                        " 	video_author varchar(20),                    "+
                        " 	video_timelong bigint,                       "+
                        " 	play_id varchar(20),                         "+
                        " 	flag bigint,                                 "+
                        " 	start_time bigint,                           "+
                        " 	end_time bigint                              "+
                        " ) WITH (                                       "+
                        "    'connector' = 'doris',                      "+
                        "    'fenodes' = 'doitedu:8030',                 "+
                        "    'table.identifier' = 'dws.video_play_agg',  "+
                        "    'username' = 'root',                        "+
                        "    'password' = 'root',                        "+
                        "    'sink.label-prefix' = 'doit42-02-love3'      "+
                        " )                                              "
        );



        // 创建 视频信息维表的 逻辑映射表
        tenv.executeSql(
                " create table video_hbase(                            "+
                        "     id int,                                  "+
                        " 	  f row<video_name string,video_type string,video_album string,video_author string,video_timelong bigint>   "+
                        " ) with (                                             "+
                        "     'connector' = 'hbase-2.2',                       "+
                        "     'table-name' = 'doit42_video',                    "+
                        "     'zookeeper.quorum' = 'doitedu:2181'              "+
                        " )                                                    "
        );

        // 对播放时长聚合结果数据，关联 视频信息维表,并将结果插入  doris 表
        tenv.executeSql(
                " INSERT INTO video_play_agg_doris                          "+
                        " SELECT                                                    "+
                        "    a.user_id,                                             "+
                        "    COALESCE(a.province,'海南省') as province,                   "+
                        "    a.device_type,                                         "+
                        "    a.video_id,                                            "+
                        "    COALESCE(v.f.video_name,'默认名称') as video_name,            "+
                        "    COALESCE(v.f.video_type,'默认类别') as video_type,            "+
                        "    COALESCE(v.f.video_author,'默认作者') as video_author,         "+
                        "    COALESCE(v.f.video_timelong,60000) as video_timelong,         "+
                        "    a.play_id,                                             "+
                        "    a.flag,                                                "+
                        "    a.start_time,                                          "+
                        "    a.end_time                                              "+
                        " FROM agg_res a                                            "+
                        " LEFT JOIN video_hbase FOR SYSTEM_TIME AS OF a.pt AS v     "+
                        " ON a.video_id = v.id                                      "
        );

    }
}
